package com.aididi.util;

import com.aididi.config.MQProducerProperties;
import com.aididi.dto.OrderDto;
import com.aididi.enums.Symbol;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @author ：xuzexiang
 * @description：MQ发送消息类
 * @date ： 2021/10/28 11:45
 */
@Configuration
@Slf4j
public class ProducerSendUtils {

    @Resource
    private DefaultMQProducer defaultMQProducer;
    @Resource
    private TransactionMQProducer transactionMQProducer;

    @Resource
    MQProducerProperties mqProducerConfig;

    /**
     * 同步发送，producer发送时会同步等待broker返回一个发送状态。如果失败会重试。吞吐量最低，但安全
     * @param topic
     * @param tag
     * @param msg
     * @return
     */
    public String send(String topic,String tag,String msg){
        log.info("发送消息到MQ:{}",msg);
        try {
            Message message = new Message(topic,tag,msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult result = defaultMQProducer.send(message);
            log.info("消息发送响应：{}",result.toString());
            return SendStatus.SEND_OK.equals(result.getSendStatus()) ? Symbol.YES.value() : Symbol.NO.value();
        } catch (Exception e) {
            log.error("消息发送失败：{}",e);
            return Symbol.NO.value();
        }
    }

    /**
     * 异步发送，producer在发送后去做自己的事情，异步接受broker的回调结果，比较有趣的地方就是引入了一个countDownLatch来保证所有消息回调方法都执行完了再关闭Producer。 所以从这里可以看出，RocketMQ的Producer也是一个服务端，在往Broker发送消息的时候也要作为服务端提供服务
     * @param topic
     * @param tag
     * @param msg
     * @return
     */
    public String sendAsync(String topic,String tag,String msg){
        log.info("发送消息到MQ:{}",msg);
        try {
            Message message = new Message(topic,tag,msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            CountDownLatch countDownLatch = new CountDownLatch(1);
            defaultMQProducer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    countDownLatch.countDown();
                    log.info(sendResult.getMsgId());
                }

                @Override
                public void onException(Throwable throwable) {
                    countDownLatch.countDown();
                    log.error(throwable.getLocalizedMessage());
                }
            });
            log.info("消息发送响应：{}");
            return Symbol.YES.value() ;
        } catch (Exception e) {
            log.error("消息发送失败：{}",e);
            return Symbol.NO.value();
        }
    }

    /**
     *  发送单条消息
     * @param topic
     * @param tag
     * @param msg
     */
    public String sendOneWay(String topic,String tag,String msg){
        log.info("发送消息到MQ:{}",msg);
        try {
            Message message = new Message(topic,tag,msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            defaultMQProducer.sendOneway(message);
            return Symbol.YES.value() ;
        } catch (Exception e) {
            log.error("消息发送失败：{}",e);
            return Symbol.NO.value();
        }
    }

    /**
     * 发送事务消息
     */
    public String sendTransationMsg(String topic,String tag,String msg){
        log.info("发送消息到MQ:{}",msg);
        //tag数组
        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
        try {
           // Message message = new Message(topic,tag,msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            //轮询的向5个Tag中发送消息，每个tag发送2个消息
            for (int i = 0; i < 10; i++) {
                try {
                    //轮询的向5个Tag中发送消息，每个tag发送2个消息
                    Message message =
                            new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult sendResult = transactionMQProducer.sendMessageInTransaction(message, null);
                    System.out.printf("%s%n", sendResult);

                    Thread.sleep(10);
                    log.info("消息发送响应：{}",sendResult.toString());
                } catch (MQClientException | UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
            return Symbol.YES.value() ;
        } catch (Exception e) {
            log.error("消息发送失败：{}",e);
            return Symbol.NO.value();
        }
    }

    /**
     * 发送有序消息
     *   producer端确保消息顺序唯一要做的事情就是将消息路由到特定的messagequeue，在RocketMQ中，通过MessageQueueSelector来实现messagequeue的选择。
     */
    public String sendOrderMsg(String topic,String tag,String orderId,String msg){
        log.info("发送消息到MQ:{}",msg);
        try {
            Message message = new Message(topic,tag,"KEY"+orderId,msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            SendResult result = defaultMQProducer.send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    // arg = orderId
                    Integer id = (Integer) o;
                    // 通过orderId 和 队列数取模运算,得到要放到哪个队列
                    int index = id % list.size();
                    return list.get(index);
                }
            },orderId);
            log.info("消息发送响应：{}",result.toString());
            return SendStatus.SEND_OK.equals(result.getSendStatus()) ? Symbol.YES.value() : Symbol.NO.value();
        } catch (Exception e) {
            log.error("消息发送失败：{}",e);
            return Symbol.NO.value();
        }
    }

    /**
     * 发送过滤消息
     *  过滤消息目的是把消息发送到Topic后，让消费端根据不同的过滤条件，消费对应的消息。
     *  按Tag过滤
     */
    public String sendFilterMsg(String topic,String tag,String msg){
        log.info("发送消息到MQ:{}",msg);
        // Tags数组
        String[] tags = new String[] {"TagA", "TagB", "TagC"};
        try {
            for (int i = 0; i < 15; i++) {
                // 消息发送到TagFilterTest主题下
                Message message = new Message(topic,
                        // 取模，轮询发送到不同的tag中
                        tags[i % tags.length],
                        msg.getBytes(RemotingHelper.DEFAULT_CHARSET));

                SendResult sendResult = defaultMQProducer.send(message);
                System.out.printf("%s%n", sendResult);
            }
            return  Symbol.YES.value();
        } catch (Exception e) {
            log.error("消息发送失败：{}",e);
            return Symbol.NO.value();
        }
    }



    /**
     * 延时消息
     * 发送消息到MQ
     * 在broker存储消息时，源码中会判断是否是延时消息，如果是的话，则会修改这个消息的Topic和QueueId等信息，消息发过来之后，
     * 会先把消息存入Schedule_Topic主题中对应的队列。然后等延迟时间到了，再转发到目标队列，推送给消费者进行消费。
     *
     * @param topic MQtopic
     * @param tag   MQ tag
     * @param msg   消息
     * @return Y:成功 N：失败
     */
    public String send(String topic, String tag, String msg, int delayLevel ) {
        log.info("发送MQ消息内容：" + msg);
        try {
            Message sendMsg = new Message(topic, tag, msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            /**
             * 设置延迟发送
             * RocketMQ不支持任意时间的延时，只支持以下几个固定的延时等级,18个固定的延迟级别
             * “1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”
             * 当级别设为3为10s生效
             */
            sendMsg.setDelayTimeLevel(delayLevel);
            // 默认3秒超时
            SendResult sendResult = defaultMQProducer.send(sendMsg);
            if(sendResult.getSendStatus() == SendStatus.SEND_OK){
                //用数据库记录日志：  更新本地消息表：消息：key, 消息状态：发送成功
                log.info("发送消息mq  key  发送sucess........");
            }else{
                //用数据库记录日志： 更新本地消息表：消息：key, 消息状态：发送失败
                log.info("发送消息mq  key  发送fail........");
            }
            log.info("消息发送响应：" + sendResult.toString());
            return SendStatus.SEND_OK.equals(sendResult.getSendStatus()) ? Symbol.YES.value() : Symbol.NO.value();
        } catch (Exception e) {
            log.error("消息发送失败：{}",e);
            return Symbol.NO.value();
        }
    }

    /**
     * 发送消息到MQ,批量消息
     * 批量消息是指将多条消息合并成一个批量消息，一次发送出去。这样的好处是可以减少网络IO，提升吞吐量。批量消息的使用是有一定限制的，这些消息应该有相同的Topic，相同的waitStoreMsgOK。而且不能是延迟消息、事务消息等
     *一个批次消息的大小不要超过1MB，实际使用时，这个1MB的限制可以稍微扩大点，实际最大的限制是4194304字节，大概4MB
     *  @param msgs  消息集合
     * @return Y:成功 N：失败
     */
    public String send(List<Message> msgs) {
        if (CollectionUtils.isEmpty(msgs)) {
            return Symbol.YES.value();
        }
        log.info("发送MQ消息内容条数：" + msgs.size());
        try {
            SendResult sendResult = defaultMQProducer.send(msgs);
            log.info("消息发送响应：" + sendResult.toString());
            return SendStatus.SEND_OK.equals(sendResult.getSendStatus()) ? Symbol.YES.value() : Symbol.NO.value();
        } catch (Exception e) {
            log.error("消息发送失败：{}",e);
            return Symbol.NO.value();
        }
    }

    /**
     * 发送消息到MQ
     *
     * @param msg 消息
     * @return Y:成功 N：失败
     */
    public String send(String msg) {
        return send(mqProducerConfig.getTopic(), mqProducerConfig.getTag(), msg);
    }


    /**
     * 发送日切流水
     * @param dto
     * @param delayLevel
     * @returnY:成功 N：失败
     */
    public String sendDayEnd(OrderDto dto, int delayLevel) {
        log.info("发送订单流水到MQ,订单消息[{}]",dto.toString() );
        return send(mqProducerConfig.getTopic(), mqProducerConfig.getTag(), JSONObject.toJSONString(dto),delayLevel);
    }

}
